package com.dpwgc.pan.service;

import java.io.BufferedReader;
import java.io.InputStreamReader;
import java.net.URI;
import java.nio.ByteBuffer;
import java.nio.CharBuffer;
import java.nio.charset.Charset;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

import com.dpwgc.pan.mapper.UserMapper;
import com.dpwgc.pan.model.User;
import org.apache.commons.lang.StringUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.BlockLocation;
import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.LocatedFileStatus;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.fs.RemoteIterator;
import org.apache.hadoop.io.IOUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;
import org.springframework.web.multipart.MultipartFile;

import javax.servlet.ServletContext;
import javax.servlet.ServletOutputStream;
import javax.servlet.http.HttpServletRequest;
import javax.servlet.http.HttpServletResponse;

/**
 * HDFS文件操作
 */
@Service
public class HdfsService {

    @Value("${hdfs.hdfsPath}")
    private String hdfsPath;

    @Value("${hdfs.hdfsName}")
    private String hdfsName;

    @Value("${hdfs.confPath}")
    private String confPath;

    @Value("${hdfs.dataLocation}")
    private String dataLocation;

    @Autowired
    UserMapper userMapper;

    //缓冲区大小
    private static final int bufferSize = 1024 * 1024 * 64;

    /**
     * 获取HDFS配置信息
     * @return
     */
    public Configuration getConfiguration() {

        Configuration conf = new Configuration();

        //加载配置
        conf.addResource(new Path(confPath + "core-site.xml"));
        conf.addResource(new Path(confPath + "hdfs-site.xml"));

        conf.set("dfs.client.use.datanode.hostname", "true");//设置按照hostName检索dataNode节点（去掉这行就变成用内网ip检索dataNode节点）
        conf.set("dfs.replication","1");//设置优先级

        conf.set("fs.defaultFS", hdfsPath);
        return conf;
    }

    /**
     * 获取HDFS文件系统对象
     * @return FileSystem
     * @throws Exception
     */
    public FileSystem getFileSystem() throws Exception {
        //在构造FileSystem对象时，将hdfsName用户名参数传递进去
        return FileSystem.get(new URI(hdfsPath), getConfiguration(), hdfsName);
    }

    /**
     * 查看文件是否已存在
     * @param path 文件路径
     * @return
     * @throws Exception
     */
    public boolean existFile(String path) throws Exception {

        if (StringUtils.isEmpty(path)) {
            return false;
        }
        FileSystem fs = getFileSystem();

        Path srcPath = new Path(path);
        return fs.exists(srcPath);
    }

    /**
     * 创建文件夹
     * @param path 文件夹路径
     * @return
     * @throws Exception
     */
    public boolean mkdirService(String path) throws Exception {

        path = dataLocation + path;

        if (StringUtils.isEmpty(path)) {
            return false;
        }
        if (existFile(path)) {
            return true;
        }
        FileSystem fs = getFileSystem();
        // 目标路径
        Path srcPath = new Path(path);
        boolean flag = fs.mkdirs(srcPath);
        fs.close();
        return flag;
    }

    /**
     * 读取当前目录信息
     * @param path 文件夹路径
     * @return
     * @throws Exception
     */
    public List<Map<String, Object>> readPathInfoService(String path) throws Exception {

        path = dataLocation + path;

        if (StringUtils.isEmpty(path)) {
            return null;
        }
        if (!existFile(path)) {
            return null;
        }
        FileSystem fs = getFileSystem();
        // 目标路径
        Path newPath = new Path(path);
        FileStatus[] statusList = fs.listStatus(newPath);
        List<Map<String, Object>> list = new ArrayList<>();
        if (null != statusList && statusList.length > 0) {

            for (FileStatus fileStatus : statusList) {

                Map<String, Object> map = new HashMap<>();

                String sp = fileStatus.toString();

                String isDirectory = (sp.split("isDirectory=")[1]).split(";")[0];//切割获取该文件类别（判断是否是目录）

                String length = "0";

                //如果是文件
                if(isDirectory.equals("false")){
                    length = (sp.split("length=")[1]).split(";")[0];//切割获取该文件大小
                }

                map.put("filePath", fileStatus.getPath());
                map.put("isDirectory", Boolean.parseBoolean(isDirectory));
                map.put("length", length);
                list.add(map);
            }
            return list;
        } else {
            return null;
        }
    }

    /**
     * 获取文件在集群中的位置
     * @param path 文件路径
     * @return
     * @throws Exception
     */
    public BlockLocation[] getFileBlockLocationsService(String path) throws Exception {

        path = dataLocation + path;

        if (StringUtils.isEmpty(path)) {
            return null;
        }
        if (!existFile(path)) {
            return null;
        }
        FileSystem fs = getFileSystem();
        //目标路径
        Path srcPath = new Path(path);
        FileStatus fileStatus = fs.getFileStatus(srcPath);
        return fs.getFileBlockLocations(fileStatus, 0, fileStatus.getLen());
    }

    /**
     * 文件上传
     * @param path 文件保存路径
     * @param file 文件
     * @throws Exception
     */
    public boolean uploadFileService(String path, MultipartFile file) throws Exception {

        path = dataLocation + path;

        String fileName = file.getOriginalFilename();

        FileSystem fs = getFileSystem();

        try{

            //拼接文件路径
            Path newPath = new Path(path + "/" + fileName);

            //打开输出流
            FSDataOutputStream outputStream = fs.create(newPath);
            outputStream.write(file.getBytes());
            outputStream.close();
            fs.close();
            return true;

        }catch (Exception e){

            fs.close();
            return false;

        } finally {
            fs.close();
        }
    }

    /**
     * 读取文件内容
     * @param path 文件路径
     * @return
     * @throws Exception
     */
    public String readFileService(String path) throws Exception {

        path = dataLocation + path;

        if (StringUtils.isEmpty(path)) {
            return null;
        }
        if (!existFile(path)) {
            return null;
        }
        FileSystem fs = getFileSystem();
        //目标路径
        Path srcPath = new Path(path);
        FSDataInputStream inputStream = null;
        try {
            inputStream = fs.open(srcPath);
            //防止中文乱码
            BufferedReader reader = new BufferedReader(new InputStreamReader(inputStream));
            String lineTxt = "";
            StringBuffer stringBuffer = new StringBuffer();
            while ((lineTxt = reader.readLine()) != null) {
                stringBuffer.append(lineTxt);
            }
            return stringBuffer.toString();
        } finally {
            inputStream.close();
            fs.close();
        }
    }

    /**
     * 文件下载（返回字符流数据给前端）
     * @param path 文件路径
     * @return
     * @throws Exception
     */
    public void downloadFileService(String path, HttpServletRequest request, HttpServletResponse response) throws Exception {

        path = dataLocation + path;

        FileSystem fs = getFileSystem();
        //目标路径
        Path srcPath = new Path(path);

        FSDataInputStream fis = fs.open(srcPath);

        //获取整个web应用，可以和程序的容器（服务器）通信
        ServletContext servletContext = request.getServletContext();
        //设置resposne响应头
        //设置响应头类型
        String mimeTyep = servletContext.getMimeType("1.jpg");
        response.setHeader("content-type",mimeTyep);
        //设置响应头打开方式
        response.setHeader("content-disposition","attachment;filename="+"file");

        //获取输出流(具体到文件名)
        ServletOutputStream out = response.getOutputStream();

        //将输入流的数据写出到输出流中
        byte[] buf = new byte[1024*8];
        int len=0;
        while ((len = fis.read(buf))!=-1){
            out.write(buf,0,len);
        }

        //关流
        IOUtils.closeStream(fis);
        fs.close();
    }

    /**
     * 获取该目录下的所有文件信息
     * @param path 文件夹路径
     * @return
     * @throws Exception
     */
    public List<Map<String, String>> listFileService(String path) throws Exception {

        path = dataLocation + path;

        if (StringUtils.isEmpty(path)) {
            return null;
        }
        if (!existFile(path)) {
            return null;
        }

        FileSystem fs = getFileSystem();
        //目标路径
        Path srcPath = new Path(path);
        //递归找到所有文件
        RemoteIterator<LocatedFileStatus> filesList = fs.listFiles(srcPath, true);

        List<Map<String, String>> returnList = new ArrayList<>();

        while (filesList.hasNext()) {
            LocatedFileStatus next = filesList.next();

            String fileName = next.getPath().getName();
            Path filePath = next.getPath();

            Map<String, String> map = new HashMap<>();
            map.put("fileName", fileName);
            map.put("filePath", filePath.toString());
            returnList.add(map);
        }

        fs.close();
        return returnList;
    }

    /**
     * 重命名文件
     * @param oldPath 旧文件路径
     * @param newPath 新文件路径
     * @return
     * @throws Exception
     */
    public boolean renameFileService(String oldPath, String newPath) throws Exception {

        oldPath = dataLocation + oldPath;
        newPath = dataLocation + newPath;

        if (StringUtils.isEmpty(oldPath) || StringUtils.isEmpty(newPath)) {
            return false;
        }
        FileSystem fs = getFileSystem();

        boolean flag = fs.rename(new Path(oldPath), new Path(newPath));
        fs.close();
        return flag;
    }

    /**
     * 删除文件
     * @param path 文件路径
     * @return
     * @throws Exception
     */
    public boolean deleteFileService(String path) throws Exception {

        path = dataLocation + path;

        if (StringUtils.isEmpty(path)) {
            return false;
        }
        if (!existFile(path)) {
            return false;
        }
        FileSystem fs = getFileSystem();
        Path srcPath = new Path(path);
        boolean flag = fs.deleteOnExit(srcPath);
        fs.close();
        return flag;
    }

    /**
     * 复制文件
     * @param sourcePath 原文件路径
     * @param targetPath 目标文件路径
     * @throws Exception
     */
    public void copyFileService(String sourcePath, String targetPath) throws Exception {

        sourcePath = dataLocation + sourcePath;
        targetPath = dataLocation + targetPath;

        if (StringUtils.isEmpty(sourcePath) || StringUtils.isEmpty(targetPath)) {
            return;
        }
        FileSystem fs = getFileSystem();
        //原文件路径
        Path oldPath = new Path(sourcePath);
        //目标文件路径
        Path newPath = new Path(targetPath);

        FSDataInputStream inputStream = null;
        FSDataOutputStream outputStream = null;
        try {
            inputStream = fs.open(oldPath);
            outputStream = fs.create(newPath);

            IOUtils.copyBytes(inputStream, outputStream, bufferSize, false);
        } finally {
            inputStream.close();
            outputStream.close();
            fs.close();
        }
    }

    /**
     * 获取该目录下的文件总数及总大小
     * @param path 文件夹路径
     * @return
     * @throws Exception
     */
    public Map<String,Object> countFileService(String path,int userId) throws Exception {

        path = dataLocation + path;

        if (StringUtils.isEmpty(path)) {
            return null;
        }
        if (!existFile(path)) {
            return null;
        }

        FileSystem fs = getFileSystem();
        //目标路径
        Path srcPath = new Path(path);
        //递归找到所有文件
        RemoteIterator<LocatedFileStatus> filesList = fs.listFiles(srcPath, true);

        int num = 0;
        long len = 0;

        while (filesList.hasNext()) {
            LocatedFileStatus next = filesList.next();

            len = len + next.getLen();

            num++;
        }

        userMapper.selectById(userId);

        //获取用户信息
        User user = userMapper.selectById(userId);

        Map<String,Object> map = new HashMap<>();
        map.put("fileNum",num);//存储文件数量
        map.put("fileLen",len);//存储文件总大小
        map.put("fileTopLen",user.getStorageSize());//用户个人存储空间上限 //1000000000

        fs.close();
        return map;
    }
}

